/**
 * 
 */
package spatial.join;

import indexing.grid.Cell;
import indexing.grid.Grid;

import java.awt.Shape;
import java.io.IOException;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
import java.util.Vector;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import spatial.geometry.MyPolygon;

/**
 * @author Xiaohui
 * @version 0.1
 * @version 0.2 Separated Mapper and Reduce TODO overlaptype is changed to a
 *          class for readability
 */
public class RasterApp extends Configured implements Tool {
	// Default constructor
	public RasterApp() {

	}

	// where to put the data in hdfs when we're done
	private static final String OUTPUT_PATH = "join_results";

	// where to read the data from.
	// File format is "id n x1...xn y1...yn"
	private static final String INPUT_PATH = "data";

	// grids size
	private static final int gridSize = 250;
	private static final double cellSize = 4;
	final static String sep = ",";

	// overlap types
	private static String[] OverlapType = { "weak", "strong", "full", "empty" };

	static class RasterAppMapper extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		Grid grids = new Grid(gridSize, cellSize);

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// construct grids
			// read polygon points
			// build polygon to IPolygon
			// grid.intersectedCells returns a set of cells
			// write the cells into <cellid, <(FULL, plg id)>
			// collect these pairs
			String coords = value.toString();
			Scanner sc = new Scanner(coords);
			int plgid = Integer.parseInt(sc.next());
			int n = Integer.parseInt(sc.next());
			if (n > 0) {
				int[] xs = new int[n];
				int[] ys = new int[n];
				for (int i = 0; i < n; i++) {
					xs[i] = Integer.parseInt(sc.next());
				}
				for (int i = 0; i < n; i++) {
					ys[i] = Integer.parseInt(sc.next());
				}
				MyPolygon plg = new MyPolygon(plgid, xs, ys);
				// TODO do logging here
//				System.out.println("Polygon created: " + plg.getID());
				// TODO very important to test the correctness of the following
				// method
				Set<Cell> cells = grids.intersectedCells((Shape) plg);
				for (Cell c : cells) {
					// check the overlapping condition here
					if (plg.contains(c)) {
						output.collect(new Text(c.getCellID().toString()),
								new Text(OverlapType[2] + sep + coords));
//						System.out.print(c.getCellID().toString());
//						System.out.println("-->");
//						System.out.println(OverlapType[2] + sep + coords);
					} else if (plg.contains(c.getCenter())) {
						// strongly overlapping here
						output.collect(new Text(c.getCellID().toString()),
								new Text(OverlapType[1] + sep + coords));
//						System.out.print(c.getCellID().toString());
//						System.out.println("-->");
//						System.out.println(OverlapType[1] + sep + coords);
					} else if (plg.intersects(c)) {
						// weakly overlapping
						output.collect(new Text(c.getCellID().toString()),
								new Text(OverlapType[0] + sep + coords));
//						System.out.print(c.getCellID().toString());
//						System.out.println("-->");
//						System.out.println(OverlapType[0] + sep + coords);
					} else {
						throw new RuntimeException(
								"intersectedCells in Grid Class is wrong: "
										+ "This intersection should not have been included! Cell:"
										+ c.getCellID() + " and polygon:"
										+ coords);
					}

				}
			}

		}
	}

	static class RasterAppReducer extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {
		final static int initSize = 1000;
		final static String plgsep = "-";

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			System.out
					.println("Enter reduce: processing key " + key.toString());
			// Since Reducer have put <cellid, <(FULL, plg id), (STRONG,
			// plg id>...>
			// We will iterate all values and get all intersected polygons
			Vector<String> v = new Vector<String>(initSize);
			while (values.hasNext()) {
				reporter.progress();
				String valueString = values.next().toString();
//				System.out.println(valueString);
				v.add(valueString);
			}
//			System.out.println("==============");
			check(v, output);
		}

		private void check(Vector<String> v, OutputCollector<Text, Text> output) {
			for (int i = 0; i < v.size(); i++) {
				Scanner sc = new Scanner(v.get(i)).useDelimiter(sep);
				String occup = sc.next();
				String plgid1 = sc.next();
				try {
					if (occup.equals(OverlapType[2])) {
						// the cell is fully occupied by the polygon
						for (int j = i + 1; j < v.size(); j++) {
							Scanner sc1 = new Scanner(v.get(j))
									.useDelimiter(sep);
							sc1.next(); // bypass the overlap type
							output.collect(new Text(""), new Text("Overlap: "
									+ plgid1 + plgsep + sc1.next()));
							sc1.close();
						}
					} else if (occup.equals(OverlapType[1])) {
						// the cell is strongly occupied by the polygon
						for (int j = i + 1; j < v.size(); j++) {
							Scanner sc1 = new Scanner(v.get(j))
									.useDelimiter(sep);
							String overlap = sc1.next();

							if (overlap.equals(OverlapType[2])
									|| overlap.equals(OverlapType[1])) {
								// the cell is fully or strongly occupied by a
								// second polygon
								output.collect(new Text(""), new Text(
										"Overlap: " + plgid1 + plgsep
												+ sc1.next()));
							} else {
								// the cell is weakly occupied by a second
								// polygon
								output.collect(new Text(""), new Text(
										"Candidate: " + plgid1 + plgsep
												+ sc1.next()));
							}
						}
					} else {
						// the cell is weakly occupied by the polygon
						for (int j = i + 1; j < v.size(); j++) {
							Scanner sc1 = new Scanner(v.get(j))
									.useDelimiter(sep);
							String overlap = sc1.next();

							if (overlap.equals(OverlapType[2])) {
								// the cell is fully occupied by a
								// second polygon
								output.collect(new Text(""), new Text(
										"Overlap: " + plgid1 + plgsep
												+ sc1.next()));
							} else {
								// the cell is weakly occupied by a second
								// polygon
								output.collect(new Text(""), new Text(
										"Candidate: " + plgid1 + plgsep
												+ sc1.next()));
							}
						}

					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}

		}
	}

	/** Driver for the actual MapReduce process */
	public int run(String[] args) throws IOException {
		JobConf conf = new JobConf(getConf(), RasterApp.class);

		FileInputFormat.addInputPath(conf, new Path(INPUT_PATH));
		FileOutputFormat.setOutputPath(conf, new Path(OUTPUT_PATH));

		conf.setMapperClass(RasterAppMapper.class);
		conf.setReducerClass(RasterAppReducer.class);


		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		conf.setMapOutputValueClass(Text.class);
		conf.setMapOutputKeyClass(Text.class);
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new RasterApp(), args);
		System.exit(ret);
	}

}
